"""
数据质量验证器
验证清洗后数据的质量和完整性
"""

import pandas as pd
import numpy as np
import logging
from typing import Dict, List, Tuple
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class DataQualityValidator:
    """数据质量验证器"""
    
    def __init__(self):
        self.validation_results = {}
        self.quality_metrics = {}
    
    def validate_data_completeness(self, df: pd.DataFrame) -> Dict:
        """验证数据完整性"""
        logger.info("开始验证数据完整性")
        
        total_records = len(df)
        completeness_results = {}
        
        for column in df.columns:
            non_null_count = df[column].notna().sum()
            null_count = df[column].isna().sum()
            completeness_rate = (non_null_count / total_records) * 100
            
            completeness_results[column] = {
                'total_records': total_records,
                'non_null_count': non_null_count,
                'null_count': null_count,
                'completeness_rate': completeness_rate,
                'is_acceptable': completeness_rate >= 95.0  # 95%以上认为可接受
            }
        
        overall_completeness = np.mean([result['completeness_rate'] for result in completeness_results.values()])
        
        self.validation_results['completeness'] = {
            'overall_completeness': overall_completeness,
            'column_details': completeness_results,
            'is_acceptable': overall_completeness >= 95.0
        }
        
        logger.info(f"数据完整性验证完成: 整体完整性 {overall_completeness:.2f}%")
        return self.validation_results['completeness']
    
    def validate_data_consistency(self, df: pd.DataFrame) -> Dict:
        """验证数据一致性"""
        logger.info("开始验证数据一致性")
        
        consistency_results = {}
        
        # 1. 验证航班代码格式
        flight_code_pattern = r'^[A-Z0-9]{3,10}$'
        valid_flight_codes = df['FlightCode'].str.match(flight_code_pattern, na=False).sum()
        flight_code_consistency = (valid_flight_codes / len(df)) * 100
        
        consistency_results['flight_codes'] = {
            'valid_count': valid_flight_codes,
            'total_count': len(df),
            'consistency_rate': flight_code_consistency,
            'is_acceptable': flight_code_consistency >= 95.0
        }
        
        # 2. 验证机场代码格式
        airport_code_pattern = r'^[A-Z]{3}$'
        valid_origin_airports = df['OriginAirportCodeIATA'].str.match(airport_code_pattern, na=False).sum()
        valid_dest_airports = df['DestinationAirportCodeIATA'].str.match(airport_code_pattern, na=False).sum()
        
        origin_consistency = (valid_origin_airports / len(df)) * 100
        dest_consistency = (valid_dest_airports / len(df)) * 100
        
        consistency_results['airport_codes'] = {
            'origin_consistency_rate': origin_consistency,
            'destination_consistency_rate': dest_consistency,
            'overall_consistency_rate': (origin_consistency + dest_consistency) / 2,
            'is_acceptable': origin_consistency >= 95.0 and dest_consistency >= 95.0
        }
        
        # 3. 验证国家代码格式
        country_code_pattern = r'^[A-Z]{2}$'
        valid_origin_countries = df['OriginCountry'].str.match(country_code_pattern, na=False).sum()
        valid_dest_countries = df['DestinationCountry'].str.match(country_code_pattern, na=False).sum()
        
        origin_country_consistency = (valid_origin_countries / len(df)) * 100
        dest_country_consistency = (valid_dest_countries / len(df)) * 100
        
        consistency_results['country_codes'] = {
            'origin_consistency_rate': origin_country_consistency,
            'destination_consistency_rate': dest_country_consistency,
            'overall_consistency_rate': (origin_country_consistency + dest_country_consistency) / 2,
            'is_acceptable': origin_country_consistency >= 95.0 and dest_country_consistency >= 95.0
        }
        
        # 4. 验证时间逻辑一致性
        time_logic_issues = (
            (df['ScheduledDepartureTime'] >= df['ScheduledArrivalTime']) |
            (df['RealDepartureTime'] >= df['RealArrivalTime']) |
            (df['ScheduledFlightDurationMinutes'] <= 0) |
            (df['ActualFlightDurationMinutes'] <= 0)
        ).sum()
        
        time_logic_consistency = ((len(df) - time_logic_issues) / len(df)) * 100
        
        consistency_results['time_logic'] = {
            'logic_issues_count': time_logic_issues,
            'consistency_rate': time_logic_consistency,
            'is_acceptable': time_logic_consistency >= 99.0
        }
        
        # 5. 验证延误时间合理性
        extreme_delays = (
            (df['DepartureDelayMinutes'].abs() > 300) |
            (df['ArrivalDelayMinutes'].abs() > 300)
        ).sum()
        
        delay_consistency = ((len(df) - extreme_delays) / len(df)) * 100
        
        consistency_results['delay_logic'] = {
            'extreme_delays_count': extreme_delays,
            'consistency_rate': delay_consistency,
            'is_acceptable': delay_consistency >= 95.0
        }
        
        overall_consistency = np.mean([
            consistency_results['flight_codes']['consistency_rate'],
            consistency_results['airport_codes']['overall_consistency_rate'],
            consistency_results['country_codes']['overall_consistency_rate'],
            consistency_results['time_logic']['consistency_rate'],
            consistency_results['delay_logic']['consistency_rate']
        ])
        
        self.validation_results['consistency'] = {
            'overall_consistency': overall_consistency,
            'details': consistency_results,
            'is_acceptable': overall_consistency >= 95.0
        }
        
        logger.info(f"数据一致性验证完成: 整体一致性 {overall_consistency:.2f}%")
        return self.validation_results['consistency']
    
    def validate_data_accuracy(self, df: pd.DataFrame) -> Dict:
        """验证数据准确性"""
        logger.info("开始验证数据准确性")
        
        accuracy_results = {}
        
        # 1. 验证航班代码唯一性
        unique_flight_codes = df['FlightCode'].nunique()
        total_flight_codes = len(df)
        flight_code_uniqueness = (unique_flight_codes / total_flight_codes) * 100
        
        accuracy_results['flight_code_uniqueness'] = {
            'unique_count': unique_flight_codes,
            'total_count': total_flight_codes,
            'uniqueness_rate': flight_code_uniqueness,
            'is_acceptable': flight_code_uniqueness >= 99.0
        }
        
        # 2. 验证机场代码有效性
        valid_airports = ['PEK', 'SHA', 'CAN', 'SZX', 'CTU', 'XIY', 'KMG', 'URC', 'TSN', 'NKG',
                         'LAX', 'JFK', 'ORD', 'DFW', 'DEN', 'SFO', 'SEA', 'LAS', 'MIA', 'ATL',
                         'LHR', 'CDG', 'FRA', 'AMS', 'MAD', 'FCO', 'BCN', 'VIE', 'ZUR', 'MUC',
                         'NRT', 'ICN', 'SIN', 'BKK', 'KUL', 'CGK', 'MNL', 'BOM', 'DEL', 'CCU',
                         'SYD', 'MEL', 'BNE', 'PER', 'ADL', 'AKL', 'WLG', 'CHC', 'DUD', 'IVC']
        
        valid_origin_airports = df['OriginAirportCodeIATA'].isin(valid_airports).sum()
        valid_dest_airports = df['DestinationAirportCodeIATA'].isin(valid_airports).sum()
        
        origin_airport_accuracy = (valid_origin_airports / len(df)) * 100
        dest_airport_accuracy = (valid_dest_airports / len(df)) * 100
        
        accuracy_results['airport_validity'] = {
            'origin_accuracy': origin_airport_accuracy,
            'destination_accuracy': dest_airport_accuracy,
            'overall_accuracy': (origin_airport_accuracy + dest_airport_accuracy) / 2,
            'is_acceptable': origin_airport_accuracy >= 80.0 and dest_airport_accuracy >= 80.0
        }
        
        # 3. 验证国家代码有效性
        valid_countries = ['CN', 'US', 'GB', 'DE', 'FR', 'IT', 'ES', 'NL', 'JP', 'KR', 
                          'SG', 'TH', 'MY', 'ID', 'PH', 'IN', 'AU', 'NZ', 'CA', 'MX',
                          'BR', 'AR', 'CL', 'CO', 'PE', 'VE', 'RU', 'TR', 'EG', 'ZA']
        
        valid_origin_countries = df['OriginCountry'].isin(valid_countries).sum()
        valid_dest_countries = df['DestinationCountry'].isin(valid_countries).sum()
        
        origin_country_accuracy = (valid_origin_countries / len(df)) * 100
        dest_country_accuracy = (valid_dest_countries / len(df)) * 100
        
        accuracy_results['country_validity'] = {
            'origin_accuracy': origin_country_accuracy,
            'destination_accuracy': dest_country_accuracy,
            'overall_accuracy': (origin_country_accuracy + dest_country_accuracy) / 2,
            'is_acceptable': origin_country_accuracy >= 90.0 and dest_country_accuracy >= 90.0
        }
        
        # 4. 验证时间范围合理性
        current_year = datetime.now().year
        valid_years = df['DepartureYear'].between(2020, current_year + 1).sum()
        year_accuracy = (valid_years / len(df)) * 100
        
        accuracy_results['time_range'] = {
            'valid_years_count': valid_years,
            'accuracy_rate': year_accuracy,
            'is_acceptable': year_accuracy >= 95.0
        }
        
        overall_accuracy = np.mean([
            accuracy_results['flight_code_uniqueness']['uniqueness_rate'],
            accuracy_results['airport_validity']['overall_accuracy'],
            accuracy_results['country_validity']['overall_accuracy'],
            accuracy_results['time_range']['accuracy_rate']
        ])
        
        self.validation_results['accuracy'] = {
            'overall_accuracy': overall_accuracy,
            'details': accuracy_results,
            'is_acceptable': overall_accuracy >= 90.0
        }
        
        logger.info(f"数据准确性验证完成: 整体准确性 {overall_accuracy:.2f}%")
        return self.validation_results['accuracy']
    
    def validate_data_distribution(self, df: pd.DataFrame) -> Dict:
        """验证数据分布"""
        logger.info("开始验证数据分布")
        
        distribution_results = {}
        
        # 1. 延误时间分布
        delay_stats = {
            'departure_delay_mean': df['DepartureDelayMinutes'].mean(),
            'departure_delay_std': df['DepartureDelayMinutes'].std(),
            'arrival_delay_mean': df['ArrivalDelayMinutes'].mean(),
            'arrival_delay_std': df['ArrivalDelayMinutes'].std(),
            'delay_rate': (df['IsDelayed'] == True).sum() / len(df) * 100,
            'severe_delay_rate': (df['IsSeverelyDelayed'] == True).sum() / len(df) * 100
        }
        
        distribution_results['delay_distribution'] = delay_stats
        
        # 2. 航线类型分布
        route_distribution = df['RouteType'].value_counts(normalize=True) * 100
        distribution_results['route_distribution'] = route_distribution.to_dict()
        
        # 3. 国内/国际航班分布
        domestic_rate = (df['IsDomestic'] == True).sum() / len(df) * 100
        distribution_results['domestic_international'] = {
            'domestic_rate': domestic_rate,
            'international_rate': 100 - domestic_rate
        }
        
        # 4. 时间分布
        hour_distribution = df['DepartureHour'].value_counts().sort_index()
        month_distribution = df['DepartureMonth'].value_counts().sort_index()
        
        distribution_results['time_distribution'] = {
            'hour_distribution': hour_distribution.to_dict(),
            'month_distribution': month_distribution.to_dict()
        }
        
        self.validation_results['distribution'] = distribution_results
        
        logger.info("数据分布验证完成")
        return distribution_results
    
    def generate_quality_report(self, df: pd.DataFrame) -> Dict:
        """生成完整的数据质量报告"""
        logger.info("开始生成数据质量报告")
        
        # 执行所有验证
        self.validate_data_completeness(df)
        self.validate_data_consistency(df)
        self.validate_data_accuracy(df)
        self.validate_data_distribution(df)
        
        # 计算整体质量分数
        completeness_score = self.validation_results['completeness']['overall_completeness']
        consistency_score = self.validation_results['consistency']['overall_consistency']
        accuracy_score = self.validation_results['accuracy']['overall_accuracy']
        
        overall_quality_score = (completeness_score + consistency_score + accuracy_score) / 3
        
        quality_report = {
            'overall_quality_score': overall_quality_score,
            'data_volume': len(df),
            'validation_results': self.validation_results,
            'quality_grade': self._get_quality_grade(overall_quality_score),
            'recommendations': self._generate_recommendations()
        }
        
        logger.info(f"数据质量报告生成完成: 整体质量分数 {overall_quality_score:.2f}")
        return quality_report
    
    def _get_quality_grade(self, score: float) -> str:
        """根据质量分数获取质量等级"""
        if score >= 95:
            return "优秀 (A)"
        elif score >= 90:
            return "良好 (B)"
        elif score >= 80:
            return "合格 (C)"
        elif score >= 70:
            return "需要改进 (D)"
        else:
            return "不合格 (F)"
    
    def _generate_recommendations(self) -> List[str]:
        """生成改进建议"""
        recommendations = []
        
        if 'completeness' in self.validation_results:
            completeness = self.validation_results['completeness']['overall_completeness']
            if completeness < 95:
                recommendations.append("建议提高数据完整性，减少缺失值")
        
        if 'consistency' in self.validation_results:
            consistency = self.validation_results['consistency']['overall_consistency']
            if consistency < 95:
                recommendations.append("建议提高数据一致性，统一数据格式")
        
        if 'accuracy' in self.validation_results:
            accuracy = self.validation_results['accuracy']['overall_accuracy']
            if accuracy < 90:
                recommendations.append("建议提高数据准确性，验证数据有效性")
        
        if not recommendations:
            recommendations.append("数据质量良好，可以用于后续分析")
        
        return recommendations
    
    def save_validation_report(self, report: Dict, filename: str = 'data_quality_report.txt'):
        """保存验证报告到文件"""
        with open(filename, 'w', encoding='utf-8') as f:
            f.write("=" * 50 + "\n")
            f.write("数据质量验证报告\n")
            f.write("=" * 50 + "\n\n")
            
            f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"数据量: {report['data_volume']:,} 条\n")
            f.write(f"整体质量分数: {report['overall_quality_score']:.2f}%\n")
            f.write(f"质量等级: {report['quality_grade']}\n\n")
            
            # 详细结果
            f.write("详细验证结果:\n")
            f.write("-" * 30 + "\n")
            
            for category, results in report['validation_results'].items():
                f.write(f"\n{category.upper()}:\n")
                if isinstance(results, dict):
                    for key, value in results.items():
                        if isinstance(value, dict):
                            f.write(f"  {key}:\n")
                            for sub_key, sub_value in value.items():
                                f.write(f"    {sub_key}: {sub_value}\n")
                        else:
                            f.write(f"  {key}: {value}\n")
            
            # 改进建议
            f.write(f"\n改进建议:\n")
            f.write("-" * 30 + "\n")
            for i, recommendation in enumerate(report['recommendations'], 1):
                f.write(f"{i}. {recommendation}\n")
        
        logger.info(f"验证报告已保存到 {filename}")

if __name__ == "__main__":
    # 测试数据验证
    validator = DataQualityValidator()
    
    # 读取清洗后的数据
    df = pd.read_csv('cleaned_flight_data.csv')
    print(f"验证数据: {len(df)} 条")
    
    # 生成质量报告
    report = validator.generate_quality_report(df)
    
    # 打印报告摘要
    print(f"\n数据质量报告摘要:")
    print(f"数据量: {report['data_volume']:,} 条")
    print(f"整体质量分数: {report['overall_quality_score']:.2f}%")
    print(f"质量等级: {report['quality_grade']}")
    
    # 保存详细报告
    validator.save_validation_report(report)
    print(f"\n详细报告已保存到 data_quality_report.txt")
